package com.uuah.server.transport.provider.mina;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.Semaphore;

import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.transport.socket.SocketConnector;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import com.uuah.exception.UuahException;
import com.uuah.server.api.IUuahEvent;
import com.uuah.server.api.IUuahMessage;
import com.uuah.server.api.endpoint.IUuahEndpoint;
import com.uuah.server.api.transport.IConnector;
import com.uuah.server.exception.transformer.TransformerException;
import com.uuah.server.message.UuahMessage;
import com.uuah.server.transport.AbstractMessageDispatcher;
import com.uuah.server.transport.provider.mina.codecs.UuahMessageCodecFactory;
import com.uuah.utils.ClassUtils;

/**
 *
 * TODO comment ...
 *
 * <p>
 * update record:updator，update time，update content and version
 * </p>
 *
 * @author <a href="jonny_quan@hotmail.com">jonny</a>
 * @date 2009 2009-3-27 下午02:45:33
 * @version 1.0.0
 */
public class MinaMessageDispatcher extends AbstractMessageDispatcher {
	protected MinaConnector connector;
	protected final SocketConnector skt;

	protected boolean useSingleThreadModel = true;
	private Semaphore lock = new Semaphore(1);

	public MinaMessageDispatcher(final IUuahEndpoint endpoint) {
		super(endpoint);
		logger.info("MinaMessageDispatcher");
		this.connector = (MinaConnector) endpoint.getConnector();
		URI uri = endpoint.getEndpointURI().getUri();

		if (this.connector.hasServer(uri)) {
			skt = null;
		} else {
			skt = new NioSocketConnector(Runtime.getRuntime()
					.availableProcessors() + 1);

			try {
				skt.getFilterChain()
						.addLast(
								"protocolFilter",
								new ProtocolCodecFilter(this
										.getProtocolCodecFactory()));
			} catch (Exception e) {
				this.exceptionThrown(e);
			}

			if (useSingleThreadModel == false) {
				Executor executor = (Executor) this.connector
						.getDispatcherThreadingProfile().createPool(
								"MinaProtocolHandler");
				skt.getFilterChain().addLast("Executor",
						new ExecutorFilter(executor));
			}

		}
	}

	@Override
	public void doDispatch(final IUuahEvent event) throws Exception {
		if (logger.isDebugEnabled()) {
			logger.debug("doDispatch");
		}
		Number id = event.getMessage().getLongProperty(
				MinaProperties.MINA_SESSION_ID_PROPERTY,
				MinaProperties.MINA_SESSION_ID_NOT_SET);
		Number[] multipleId = (Number[]) event.getMessage().getProperty(
				MinaProperties.MINA_MULTICAST_SESSION_IDS_PROPERTY);

		URI endpointUri = event.getEndpoint().getEndpointURI().getUri();

		if (this.connector.hasServer(endpointUri)) {
			if (multipleId == null) {
				sendTo(id, endpointUri, event);
			} else {
				for (Number sessionId : multipleId) {
					sendTo(sessionId, endpointUri, event);
				}
			}
		} else {
			this.sendToRemote(event);

		}
	}

	private void sendTo(Number sessionId, URI endpointUri, final IUuahEvent event)
			throws TransformerException {
		IoSession session = this.connector.getSession(endpointUri, sessionId);
		if (session == null) {
			logger.warn("SESSION NOT FOUND: " + sessionId + " ENDPOINT: "
					+ endpoint);
		} else {
			Object payload = this.getPayload(event);
			if (payload != null) {
				session.write(payload);
			}
		}
	}

	@Override
	public IUuahMessage doSend(final IUuahEvent event) throws Exception {
		if (logger.isDebugEnabled()) {
			logger.debug("onSend");
		}
		URI endpointUri = event.getEndpoint().getEndpointURI().getUri();
		if (this.connector.hasServer(endpointUri)) {
			// Send a response to a connected client:
			Number id = event.getMessage().getLongProperty(
					MinaProperties.MINA_SESSION_ID_PROPERTY,
					IConnector.INT_VALUE_NOT_SET);

			IoSession session = this.connector.getSession(endpointUri, id);
			if (session == null) {
				logger.warn("SESSION NOT FOUND: " + id + " ENDPOINT: "
						+ endpoint);
			} else {
				Object payload = this.getPayload(event);
				if (payload != null) {
					WriteFuture write = session.write(payload);
				}
			}
			return null;
		} else {
			this.lock.acquire();
			ConnectFuture future = this.sendToRemote(event);
			this.lock.acquire();

			IoSession session = future.getSession();
			Object result = session
					.getAttribute(MinaProperties.LAST_MESSAGE_RECEIVED_ATTRIBUTE);
			if (null == result) {
				return null;
			} else {
				return new UuahMessage(connector.getMessageAdapter(result));
			}

		}
	}

	public Object getDelegateSession() throws UuahException {
		return null;
	}

	@Override
	protected void doConnect() throws Exception {
	}

	@Override
	public void doDispose() {
	}

	@Override
	protected void doDisconnect() throws Exception {
	}

	/**
	 * Connect to a host and send the payload
	 *
	 * todo:We should have the possibility to keep the connection open and send
	 * more request
	 *
	 * @param event
	 *            The UMOEvent generating the request
	 * @return a future of the connection attemp
	 * @throws Exception
	 */
	protected ConnectFuture sendToRemote(IUuahEvent event) throws Exception {
		URI endpointUri = event.getEndpoint().getEndpointURI().getUri();

		int port = endpointUri.getPort();
		port = (port == -1) ? 80 : port;

		InetSocketAddress address = new InetSocketAddress(
				endpointUri.getHost(), port);
		IoHandler clientHandler = this.getClientHandler(event, this
				.getPayload(event), this.lock);
		skt.setHandler(clientHandler);
		ConnectFuture cFuture = skt.connect(address);
		return cFuture;
	}

	protected Object getPayload(IUuahEvent event) throws TransformerException {
		return event.getMessage().getPayload();
	}

	protected IoHandler getClientHandler(IUuahEvent event, Object message,
			Semaphore lock) throws UuahException {
		return new MinaDefalutClientHandler(connector, event.getService()
				.getComponent(), endpoint, event, message, lock);
	}

	@SuppressWarnings("unchecked")
	protected ProtocolCodecFactory getProtocolCodecFactory() throws Exception {
		Map<String, String> props = endpoint.getProperties();
		String protocolFactoryClassName = (String) props.get("protocolFactory");
		if (protocolFactoryClassName == null) {
			if (logger.isDebugEnabled()) {
				logger.debug("Using default protocol codecs");
			}
			protocolFactoryClassName = UuahMessageCodecFactory.class.getName();
		}

		return (ProtocolCodecFactory) ClassUtils.instanciateClass(
				protocolFactoryClassName, new Object[] { false });
	}
}
